package com.sh.data.engine.domain.integration.api.taskconfig.service.impl;

import cn.hutool.core.convert.Convert;
import cn.hutool.core.lang.Snowflake;
import cn.hutool.core.lang.UUID;
import com.alibaba.fastjson.JSON;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.Lists;
import com.jayway.jsonpath.Configuration;
import com.jayway.jsonpath.JsonPath;
import com.sh.data.engine.common.constants.Constants;
import com.sh.data.engine.common.enumDefinition.DSType;
import com.sh.data.engine.common.enumDefinition.OfflineSyncTaskStatus;
import com.sh.data.engine.common.exception.BusinessException;
import com.sh.data.engine.common.util.ServletUtils;
import com.sh.data.engine.domain.authority.service.DataAuthService;
import com.sh.data.engine.domain.base.model.PageResult;
import com.sh.data.engine.domain.base.model.UserContext;
import com.sh.data.engine.domain.base.model.datax.DataX;
import com.sh.data.engine.domain.base.model.datax.JobWriterConnection;
import com.sh.data.engine.domain.base.model.enums.RightEnum;
import com.sh.data.engine.domain.base.service.LogService;
import com.sh.data.engine.domain.dataasset.model.domain.TableInfoBizDomain;
import com.sh.data.engine.domain.dataasset.service.TableService;
import com.sh.data.engine.domain.dataasset.service.TblPropService;
import com.sh.data.engine.domain.datadev.offline.model.enums.OfflineDevJobStatusEnum;
import com.sh.data.engine.domain.integration.api.record.model.domain.ApiRunningRecordDomain;
import com.sh.data.engine.domain.integration.api.record.service.ApiRunningRecordService;
import com.sh.data.engine.domain.integration.api.taskconfig.model.domain.ApiLogDomain;
import com.sh.data.engine.domain.integration.api.taskconfig.model.domain.JarUploadResultDomain;
import com.sh.data.engine.domain.integration.api.taskconfig.model.domain.TaskRecordDomain;
import com.sh.data.engine.domain.integration.api.taskconfig.model.param.ExpressionParam;
import com.sh.data.engine.domain.integration.api.taskconfig.model.param.PreviewExpressParam;
import com.sh.data.engine.domain.integration.api.taskconfig.model.param.TaskConfigParam;
import com.sh.data.engine.domain.integration.api.taskconfig.model.param.TaskRecordParam;
import com.sh.data.engine.domain.integration.api.taskconfig.service.TaskConfigService;
import com.sh.data.engine.domain.integration.datasource.model.domain.DataSourceDetailDomain;
import com.sh.data.engine.domain.integration.datasource.model.domain.DataSourceDomain;
import com.sh.data.engine.domain.integration.datasource.model.domain.DataSourceManagerDomain;
import com.sh.data.engine.domain.integration.datasource.service.DataSourceService;
import com.sh.data.engine.domain.integration.offline.model.domain.OfflineSyncDomain;
import com.sh.data.engine.domain.integration.offline.service.DataXService;
import com.sh.data.engine.domain.shims.db.BaseDbManager;
import com.sh.data.engine.domain.shims.db.model.FieldInfoDomain;
import com.sh.data.engine.domain.shims.db.model.TableInfoDomain;
import com.sh.data.engine.domain.shims.hbase.manager.HBaseManager;
import com.sh.data.engine.domain.shims.mongo.manager.MongoManager;
import com.sh.data.engine.domain.shims.sql.builder.DDLBuilder;
import com.sh.data.engine.domain.shims.sql.ddl.CreateTableParam;
import com.sh.data.engine.domain.shims.sql.ddl.DDL;
import com.sh.data.engine.domain.workspace.index.service.ProjectService;
import com.sh.data.engine.domain.workspace.manager.service.impl.StorageServiceImpl;
import com.sh.data.engine.infrastructure.config.DataXConfig;
import com.sh.data.engine.infrastructure.config.FileStorageConfiguration;
import com.sh.data.engine.repository.dao.dataasset.entity.TblPropEntity;
import com.sh.data.engine.repository.dao.integration.api.record.entity.ApiRunningRecordEntity;
import com.sh.data.engine.repository.dao.integration.datasource.entity.DataSourceEntity;
import com.sh.data.engine.repository.dao.integration.datasource.entity.DataSourceEntity.RdbmsConfig;
import com.sh.data.engine.repository.dao.workspace.manager.entity.StorageEntity;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.commons.lang3.StringUtils;
import org.openjdk.nashorn.api.scripting.ScriptObjectMirror;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;
import org.springframework.web.multipart.MultipartFile;

import javax.script.Invocable;
import javax.script.ScriptEngine;
import javax.script.ScriptEngineManager;
import javax.script.ScriptException;
import java.io.File;
import java.io.IOException;
import java.sql.SQLException;
import java.sql.Types;
import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * @author : changkong
 * @date : 2022/3/10
 */
@Slf4j
@Service
public class TaskConfigServiceImpl implements TaskConfigService {

    @Value("${datadev.api.jar.dir:/data/upload/jars/datadev}")
    private String apiJarFileDir;

    private Snowflake snowflake = new Snowflake();

    private static final int PREVIEW_LIMIT = 6;

    @Lazy
    @Autowired
    private DataSourceService dataSourceService;

    @Autowired
    private ProjectService projectService;

    @Autowired
    private StorageServiceImpl storageServiceImpl;

    @Autowired
    private ApiRunningRecordService apiRunningRecordService;

    @Autowired
    private DataXService dataXService;

    @Autowired
    private DataXConfig dataXConfig;

    @Autowired
    private FileStorageConfiguration fileStorageConfiguration;

    @Autowired
    private LogService logService;

    @Autowired
    private DataAuthService dataAuthService;
    @Autowired
    private TableService tableService;

//    @Autowired
//    private SecretMgtService secretMgtService;

    @Autowired
    private TblPropService tblPropService;

    /**
     * 上传jar包
     *
     * @param requestFile
     * @return
     */
    @Override
    public JarUploadResultDomain uploadJar(MultipartFile requestFile) {
        String originalFilename = requestFile.getOriginalFilename();
        // 如果文件不是以.jar结尾
        if (StringUtils.isNotEmpty(originalFilename)
            && (!originalFilename.toLowerCase().endsWith(".jar"))) {
            throw new BusinessException("文件格式错误,必须是jar文件");
        }
        byte[] data;
        try {
            data = requestFile.getBytes();
        } catch (IOException e) {
            throw new BusinessException("文件格式错误");
        }

        String filePath = apiJarFileDir + File.separator + UUID.fastUUID() + ".jar";

        File file = new File(filePath);

        try {
            FileUtils.writeByteArrayToFile(file, data);
        } catch (IOException e) {
            throw new BusinessException("上传失败");
        }
        JarUploadResultDomain jarUploadResultDomain = new JarUploadResultDomain();

        String redisKey = "api:jar_" + snowflake.nextIdStr();
        jarUploadResultDomain.setFileName(originalFilename);
        jarUploadResultDomain.setFilePath(filePath);
        jarUploadResultDomain.setCacheKey(redisKey);
        return jarUploadResultDomain;
    }

    /**
     * 预览
     *
     * @param param
     * @return
     */
    @Override
    public List<List<Object>> preview(PreviewExpressParam param) {
        List<List<Object>> result;
        if (CollectionUtils.isNotEmpty(param.getExpressionParamList())) {
            result = parseJsonPath(param.getResponseRawJson(), param.getExpressionParamList());
        } else {
            result = parseJsScript(param.getResponseRawJson(), param.getJsScript());
        }
        return CollectionUtils.size(result) > PREVIEW_LIMIT ? result.subList(0, PREVIEW_LIMIT) : result;
    }

    /**
     * 获取数据目标(datasource)
     *
     * @param projectId
     * @return
     */
    @Override
    public List<DataSourceDomain> getDatabases(Long projectId) {
        return dataSourceService.getAllByProject(projectId);
    }

    /**
     * 获取数据源中的表
     *
     * @param dsId
     * @return
     */
    @Override
    public List<TableInfoDomain> getTargetTables(Long dsId) {
        DataSourceManagerDomain managerInfo = dataSourceService.getManagerInfo(dsId);
        if (Objects.isNull(managerInfo)) {
            throw new BusinessException("获取数据源信息失败，请检查后重试！");
        }
        List<TableInfoDomain> tableList = null;

        BaseDbManager baseDbManager = managerInfo.getDbManager();
        MongoManager mongoManager = managerInfo.getMongoManager();
        HBaseManager hBaseManager = managerInfo.getHBaseManager();
        String dbName = managerInfo.getDbName();
        try {
            if (!Objects.isNull(baseDbManager)) { // base
                if (baseDbManager.hasSchema()) {
                    DataSourceEntity byId = dataSourceService.getById(dsId);
                    String schema = byId.getRdbmsConfig().getSchema();
                    tableList = baseDbManager.getTableAndViewList(dbName, schema);
                } else {
                    tableList = baseDbManager.getTableAndViewList(dbName);
                }

            } else if (!Objects.isNull(mongoManager) || !Objects.isNull(hBaseManager)) { // mongo or hbase
                List<String> collectionNames = null;
                if (!Objects.isNull(mongoManager)) {
                    collectionNames = mongoManager.getCollectionList();
                } else {
                    collectionNames = hBaseManager.listCollectionNames();
                }
                tableList =
                    collectionNames.stream()
                        .map(
                            c -> {
                                TableInfoDomain tableInfo = new TableInfoDomain();
                                tableInfo.setTableName(c);
                                tableInfo.setTblType(TableInfoDomain.TblType.MANAGED_TABLE);
                                return tableInfo;
                            })
                        .collect(Collectors.toList());
            }
        } catch (Exception e) {
            log.error("get table list error {}", e.getMessage());
            throw new BusinessException("获取数据源信息失败，请检查后重试！");
        } finally {
            if (null != baseDbManager) {
                baseDbManager.close();
            }
        }

        Integer isSystem = managerInfo.getIsSystem();
        if (Objects.nonNull(isSystem) && isSystem.equals(1)) {
            String userId = UserContext.getUserId();
            Long projectId = ServletUtils.getProjectId();
            Iterator<TableInfoDomain> iterator = tableList.iterator();
            while (iterator.hasNext()) {
                TableInfoDomain next = iterator.next();
                String tableName = next.getTableName();
                TableInfoBizDomain tableBizInfo = tableService.getTableBizInfo(projectId, tableName);
                if (tableBizInfo == null) {
                    continue;
                }
                String tblAuth = dataAuthService.getTblAuth(projectId, userId, tableBizInfo.getTblId());
                if (StringUtils.isBlank(tblAuth)) {
                    continue;
                }
                // 目标：无写权限，过滤掉
                if (!RightEnum.RW.getCode().equals(tblAuth)) {
                    iterator.remove();
                }
            }
        }
        return tableList;
    }

    /**
     * 获取建表语句
     *
     * @param projectId
     * @param fieldNames 预览中设置的参数
     * @return
     */
    @Override
    public String getCreateTableDDL(
        Long projectId, Long dsId, List<String> fieldNames, Integer isSupportRtf) {
        log.info("getCreateTableDDL projectId:{}, dsId:{}, fieldNames:{}", projectId, dsId, fieldNames);
        // storageEntity 和 datasource中的配置必须是一致的
        String dsSchema = "";
        DataSourceDetailDomain dataSourceDetailById =
            dataSourceService.getDataSourceDetailById(dsId, false);
        if (null == dataSourceDetailById) {
            throw new BusinessException("目标数据源不存在");
        }
        if (DSType.PostgreSQL == DSType.from(dataSourceDetailById.getDsType())
            || DSType.Sap == DSType.from(dataSourceDetailById.getDsType())
            || DSType.KINGBASE8 == DSType.from(dataSourceDetailById.getDsType())) {
            DataSourceDetailDomain.RdbmsConfig rdbmsConfig = dataSourceDetailById.getRdbmsConfig();
            if (null == rdbmsConfig || StringUtils.isBlank(rdbmsConfig.getSchema())) {
                throw new BusinessException("datasouce中未指定schema");
            }
            dsSchema = rdbmsConfig.getSchema();
        }

        BaseDbManager baseDbManager = projectService.getDbManager(projectId);
        StorageEntity storageEntity = storageServiceImpl.tmpGetStorageInfoByProjectId(projectId);
        if (null == baseDbManager || null == storageEntity) {
            throw new BusinessException("建表语句执行失败,检查项目数据源状态");
        }
        // 根据存储类型获取对应的表操作
        DSType dsType = DSType.from(storageEntity.getStorageType());
        try (DDLBuilder ddlBuilder = baseDbManager.getDDLBuilder()) {
            CreateTableParam createTableParam = new CreateTableParam();
            if (DSType.isRdbms(storageEntity.getStorageType())) {
                StorageEntity.Rmdbs rmdbs = storageEntity.getRmdbs();
                if (null == rmdbs) {
                    throw new BusinessException("建表语句执行失败,检查项目数据源配置信息");
                }
                String schema = Optional.ofNullable(rmdbs.getSchema()).orElse("");

                // 没有schema的数据库不去比对
                if (StringUtils.isEmpty(dsSchema) && !Objects.equals(dsSchema, schema)) {
                    throw new BusinessException("检查数据源与默认存储配置的schema是否一致");
                }
                if (StringUtils.isNotEmpty(schema)) {
                    log.info("schema = {}", schema);
                    createTableParam.setSchema(schema);
                }
            }
            String tableName = "ods_" + RandomStringUtils.randomAlphabetic(6).toLowerCase();
            createTableParam.setTableName(tableName);
            createTableParam.setDbName(dataSourceDetailById.getDbName());
            if (dsType == DSType.Hive) {
                createTableParam.setStoreType(isSupportRtf == 1 ? "ORC" : "TEXTFILE");
            }
            ArrayList<CreateTableParam.FieldInfo> fields = Lists.newArrayList();
            for (String fieldName : fieldNames) {
                CreateTableParam.FieldInfo fieldInfo = new CreateTableParam.FieldInfo();
                fieldInfo.setFieldName(fieldName);
                // 默认给字符串类型
                fieldInfo.setFieldType(baseDbManager.toDbType(Types.VARCHAR));
                fields.add(fieldInfo);
            }
            createTableParam.setFieldInfoList(fields);
            DDL ddl = ddlBuilder.create(createTableParam);
            String ddlSql = ddl.toString();
            //      if (StringUtils.trim(ddlSql).endsWith(";")) {
            //        ddlSql = ddlSql.substring(0, ddlSql.lastIndexOf(";"));
            //      }
            return ddlSql;
        } catch (SQLException e) {
            log.error("创建建表语句失败{}", e.getMessage());
            throw new BusinessException("创建建表语句失败, 检查数据源状态");
        } finally {
            baseDbManager.close();
        }
    }

    /**
     * 建表
     *
     * @param projectId
     * @param sql       提交的sql语句, 需要判断是否是create语句,提取出表名
     */
    @Override
    public String saveTargetTable(Long projectId, String sql) {

        DataSourceEntity systemDataSourceDetail =
            dataSourceService.getSystemDataSourceDetail(projectId);
        if (null == systemDataSourceDetail) {
            throw new BusinessException("检查数据源状态");
        }
        return dataSourceService
            .executeCreateTableSql(systemDataSourceDetail.getId(), sql)
            .getTableName();
    }

    @Override
    public String saveTargetTable(Long projectId, String sql, Long dataSecretId) {
        DataSourceEntity systemDataSourceDetail =
            dataSourceService.getSystemDataSourceDetail(projectId);
        if (null == systemDataSourceDetail) {
            throw new BusinessException("检查数据源状态");
        }
        boolean requireSetDataSecret = Objects.nonNull(dataSecretId);
        long secretId = -1;
//        if (requireSetDataSecret) {
//            secretId = Long.valueOf(dataSecretId);
//            Long securityLevelId = UserContext.getUserInfo().getSecurityLevelId();
//            log.info("user secretId: {}, data secretId: {}", securityLevelId, secretId);
//            // 判断当前用户密级
//            if (!secretMgtService.validSecretLevel(securityLevelId, secretId)) {
//                throw new BusinessException("数据密级设置异常");
//            }
//        }
        String schemaTableName = dataSourceService
            .executeCreateTableSql(systemDataSourceDetail.getId(), sql, dataSecretId)
            .getTableName();

        TblPropEntity tblPropEntity = new TblPropEntity();
        tblPropEntity.setProjectId(projectId);
        RdbmsConfig rdbmsConfig = systemDataSourceDetail.getRdbmsConfig();
        if (Objects.nonNull(rdbmsConfig)) {
            tblPropEntity.setSchema(rdbmsConfig.getSchema());
        }
        tblPropEntity.setTblId(tblPropService.generateTblId(projectId));
        tblPropEntity.setDataSecretId(secretId);
        String tableName = null;
        if (schemaTableName.contains(".")) {
            String[] split = schemaTableName.split("\\.", 2);
            tableName = split[1];
        }
        tblPropEntity.setTableName(tableName);
        tblPropEntity.setLifecycle(-1);
        tblPropEntity.setTblType(TableInfoDomain.TblType.MANAGED_TABLE.name());
        tblPropEntity.setRowState(1);
        tblPropService.save(tblPropEntity);
        return schemaTableName;
    }


    /**
     * 获取上一步选择表的字段, 并进行排序
     *
     * @param dsId
     * @param tableName
     * @return
     */
    @Override
    public List<FieldInfoDomain> getTargetTableFields(Long dsId, String tableName) {
        List<FieldInfoDomain> fieldList = dataSourceService.getFieldList(dsId, tableName);
        List<FieldInfoDomain> sortedFieldList =
            fieldList.stream()
                .sorted(Comparator.comparing(FieldInfoDomain::getFieldName))
                .collect(Collectors.toList());
        return sortedFieldList;
    }

    /**
     * 构建datax执行脚本
     *
     * @param reader
     * @param task
     * @return
     */
    @Override
    public String buildJobJson(DataX.JobContentReader reader, TaskConfigParam task) {
        // 1.初始化datax
        DataX dataX = new DataX();
        // 2.初始化 job
        DataX.Job job = new DataX.Job();
        // 2.1 初始化setting
        DataX.JobSetting setting = new DataX.JobSetting();
        DataX.JobSettingSpeed speed = new DataX.JobSettingSpeed();
        speed.setChannel(dataXConfig.getChannelNum());
        setting.setSpeed(speed);
        if (task.getErrorNumLimit() != null
            && task.getErrorNumLimit() != 0
            && task.getErrorNumLimit() != -1) {
            DataX.JobSettingErrorLimit errorLimit = new DataX.JobSettingErrorLimit();
            errorLimit.setRecord(task.getErrorNumLimit());
            setting.setErrorLimit(errorLimit);
        }
        job.setSetting(setting);
        // 2.2 初始化 jobcontent
        List<DataX.JobContent> content = Lists.newArrayList();
        content.add(new DataX.JobContent());
        job.setContent(content);

        content.get(0).setReader(reader);

        // 构建 4、writer
        // 设置连接
        List<JobWriterConnection> writeConnections = Lists.newArrayList();
        // 关系型数据库
        String targetDsType = task.getTargetDsType();
        Integer ruleId = task.getDataReplaceRuleId();
        Long targetDsId = task.getTargetDsId();
        String targetTable = task.getTargetTableName();
        Integer needTruncate = task.getIsTruncate();

        List<TaskConfigParam.FieldMapping> taskFieldMappings = task.getFieldMappings();
        List<OfflineSyncDomain.FieldMapping> fieldMappings = new ArrayList<>();
        for (TaskConfigParam.FieldMapping taskField : taskFieldMappings) {
            OfflineSyncDomain.FieldMapping fieldMapping = new OfflineSyncDomain.FieldMapping();
            fieldMapping.setSourceField(taskField.getSourceField());
            fieldMapping.setSourceFieldType(taskField.getSourceFieldType());
            fieldMapping.setIsSourceFieldPk(0);
            fieldMapping.setTargetField(taskField.getTargetField());
            fieldMapping.setTargetFieldType(taskField.getTargetFieldType());
            fieldMapping.setIsTargetFieldPk(0);
            //      fieldMapping.setTargetColumnSize();
            //      fieldMapping.setTargetDecimalDigits();
            //      fieldMapping.setDateFormat();
            fieldMappings.add(fieldMapping);
        }

        List<TaskConfigParam.PartitionField> taskPartitions = task.getPartitions();
        List<OfflineSyncDomain.PartitionField> paritions = new ArrayList<>();
        for (TaskConfigParam.PartitionField taskPartitionField : taskPartitions) {
            OfflineSyncDomain.PartitionField partitionField = new OfflineSyncDomain.PartitionField();
            partitionField.setFieldName(taskPartitionField.getFieldName());
            partitionField.setFieldValue(taskPartitionField.getFieldValue());
            paritions.add(partitionField);
        }

        List<DataX.JobContent> jobContents =
            dataXService.buildWriter(
                content,
                targetDsType,
                ruleId,
                targetDsId,
                targetTable,
                needTruncate,
                fieldMappings,
                paritions,
                null,
                dataXConfig.getBatchSize());
        job.setContent(jobContents);
        dataX.setJob(job);

        return JSON.toJSONString(dataX, true);
    }

    /**
     * 由具体的接口采集实现类去实现
     *
     * @param taskId
     * @param type
     */
    @Override
    public void operate(Long taskId, String type) {
        throw new UnsupportedOperationException();
    }

    /**
     * 生成xxl执行代码
     *
     * @param id
     * @param beanName
     * @return
     */
    public static String generateGlueSource(String id, String beanName) {
        char[] chars = beanName.toCharArray();
        chars[0] = Character.toLowerCase(chars[0]);
        beanName = new String(chars);

        String glueJavaFormat =
            "package com.xuelangyun.hufu.job.jobhandler.api;\n"
                + "\n"
                + "import com.xuelangyun.hufu.job.core.context.XxlJobHelper;\n"
                + "import com.xuelangyun.hufu.job.core.handler.IJobHandler;\n"
                + "import com.xuelangyun.hufu.service.integration.api.taskconfig.TaskConfigService;\n"
                + "import org.springframework.beans.factory.annotation.Autowired;\n"
                + "import javax.annotation.Resource;\n"
                + "\n"
                + "public class ApiSyncDemoHandler extends IJobHandler {\n"
                + "\n"
                + "  // api同步id\n"
                + "  private String apiSyncId = \"%s\";\n"
                + "  @Resource(name=\"%s\") private TaskConfigService taskConfigService;\n"
                + "\n"
                + "  @Override\n"
                + "  public void execute() throws Exception {\n"
                + "    taskConfigService.executeApi(apiSyncId);\n"
                + "  }\n"
                + "\n"
                + "  @Override\n"
                + "  public void init(){\n"
                + "    XxlJobHelper.log(\"开始\");\n"
                + "  }\n"
                + "\n"
                + "  @Override\n"
                + "  public void destroy(){\n"
                + "    XxlJobHelper.log(\"执行人工停止操作\");\n"
                + "    taskConfigService.destroy(apiSyncId);\n"
                + "  }\n"
                + "}\n";
        return String.format(glueJavaFormat, id, beanName);
    }

    /**
     * task running record
     *
     * @param taskRecordParam
     * @return
     */
    @Override
    public PageResult<TaskRecordDomain> historyTask(TaskRecordParam taskRecordParam) {
        PageResult<ApiRunningRecordEntity> pageEntity =
            apiRunningRecordService.getRunningRecordPageList(
                taskRecordParam.getPageNum(),
                taskRecordParam.getPageSize(),
                taskRecordParam.getTaskId(),
                taskRecordParam.getTaskStatus(),
                taskRecordParam.getExecuteType(),
                taskRecordParam.getIgnored());
        List<ApiRunningRecordEntity> entityList = pageEntity.getResult();
        PageResult<TaskRecordDomain> pageDomain = Convert.convert(PageResult.class, pageEntity);
        List<TaskRecordDomain> domainList = Convert.toList(TaskRecordDomain.class, entityList);
        domainList.stream()
            .forEach(
                d -> {
                    d.setTaskName(taskRecordParam.getTaskName());
                    d.setCreator(taskRecordParam.getCreateBy());
                    if (!Objects.isNull(d.getEndTime()) && !Objects.isNull(d.getStartTime())) {
                        long l = (d.getEndTime().getTime() - d.getStartTime().getTime()) / 1000;
                        d.setDuration(l);
                    }
                });

        pageDomain.setResult(domainList);
        return pageDomain;
    }

    /**
     * 如果设置了schema, 生成datax json时需要加上
     *
     * @param task
     * @return
     */
    public static String dataXTaskAddSchema(
        TaskConfigParam task, DataSourceDetailDomain dataSourceDetailDomain) {
        String targetDsType = task.getTargetDsType();
        if (null == dataSourceDetailDomain) {
            throw new BusinessException("目标数据源不存在");
        }
        // 原tablename
        String originTableName = task.getTargetTableName();
        if (targetDsType.equalsIgnoreCase(DSType.PostgreSQL.name())) {
            DataSourceDetailDomain.RdbmsConfig rdbmsConfig = dataSourceDetailDomain.getRdbmsConfig();
            if (null == rdbmsConfig || StringUtils.isBlank(rdbmsConfig.getSchema())) {
                throw new BusinessException("Pg未指定schema");
            }
            return rdbmsConfig.getSchema() + "." + originTableName;
        }
        return originTableName;
    }

    /**
     * @param json
     * @param jsScript
     * @return
     */
    private static List<List<Object>> parseJsScript(String json, String jsScript) {
        ScriptEngineManager scriptEngineManager = new ScriptEngineManager();
        ScriptEngine engine = scriptEngineManager.getEngineByName("javascript");
        // 执行js脚本定义函数
        Object res = null;
        try {
            engine.eval(jsScript);
            Invocable invocable = (Invocable) engine;
            res = invocable.invokeFunction("parseJson", json);
        } catch (ScriptException e) {
            throw new BusinessException(e.getMessage());
        } catch (NoSuchMethodException e) {
            throw new BusinessException("函数名必须定义:parseJson");
        }
        List<List<Object>> lists = parseJsObj(res, false);
        return lists;
    }

    /**
     * 解析执行js脚本返回的对象
     *
     * @param object
     * @return
     */
    private static List<List<Object>> parseJsObj(Object object, boolean previewFlag) {
        AtomicInteger count = new AtomicInteger();
        ScriptObjectMirror sjm = (ScriptObjectMirror) object;
        List<List<Object>> result = Lists.newArrayList();
        for (Map.Entry<String, Object> entry : sjm.entrySet()) {
            Object j = entry.getValue();
            ScriptObjectMirror inner = (ScriptObjectMirror) j;
            List<Object> node = Lists.newArrayList();
            count.getAndIncrement();
            inner.forEach(
                (m, n) -> {
                    Object o = JSON.toJSON(n);
                    if (o instanceof String) {
                        String s = o.toString();
                        if (s.startsWith("\"") && s.endsWith("\"")) {
                            s = s.substring(1, s.length() - 1);
                        }
                        o = s;
                    }
                    node.add(o);
                });
            result.add(node);
            if (previewFlag) {
                if (count.get() > 5) {
                    break;
                }
            }
        }
        return result;
    }

    public static void main(String[] args) {
        String jsScript = "function parseJson(inputParam) {\n" +
            "  // inputParam为返回报文的JSON字符串\n" +
            "  var dataObj = JSON.parse(inputParam)\n" +
            "  var result=[['name','value','time']]  // 定义参数名\n" +
            "  for(i=0; i < dataObj.values.length; i++) {  // 此处输入处理逻辑,以循环遍历为例\n" +
            "      var arr = []\n" +
            "      var name = dataObj.values[i].name  // 输入叶子节点数据\n" +
            "      var value = dataObj.values[i].value  // 输入叶子节点数据\n" +
            "      var time = dataObj.values[i].properties.updatetime  // 输入叶子节点数据\n" +
            "      time = time.replace(' ', 'T') + 'Z'\n" +
            "      time = new Date(time).getTime() + ''\n" +
            "      arr.push(\"abc\")\n" +
            "      arr.push(345.12)\n" +
            "      arr.push(time)\n" +
            "      result.push(arr)\n" +
            "  }\n" +
            "   return result\n" +
            "}";
        String json = "{\n" +
            "  \"values\": [\n" +
            "    {\n" +
            "      \"name\": \"name1\",\n" +
            "      \"value\": 1,\n" +
            "      \"properties\": {\n" +
            "        \"updatetime\": \"2021-01-01 00:00:00\"\n" +
            "      }\n" +
            "    },\n" +
            "    {\n" +
            "      \"name\": \"name2\",\n" +
            "      \"value\": 2,\n" +
            "      \"properties\": {\n" +
            "        \"updatetime\": \"2021-01-02 00:00:00\"\n" +
            "      }\n" +
            "    },\n" +
            "    {\n" +
            "      \"name\": \"name3\",\n" +
            "      \"value\": 3,\n" +
            "      \"properties\": {\n" +
            "        \"updatetime\": \"2021-01-03 00:00:00\"\n" +
            "      }\n" +
            "    }\n" +
            "  ]\n" +
            "}";

        List<List<Object>> lists = parseJsScript(json, jsScript);
        System.out.println(lists);
    }

    /**
     * 解析jsonPath
     *
     * @param json
     * @param expressionParamList
     * @return
     */
    private List<List<Object>> parseJsonPath(String json, List<ExpressionParam> expressionParamList) {
        Object document = Configuration.defaultConfiguration().jsonProvider().parse(json);

        ObjectMapper objectMapper = new ObjectMapper();

        Set<String> names =
            expressionParamList.stream().map(ExpressionParam::getName).collect(Collectors.toSet());

        if (!Objects.equals(names.size(), expressionParamList.size())) {
            throw new BusinessException("jsonPath表达式名称不能重复");
        }

        // 父级
        List<Object> parentList = new ArrayList<>();
        List<ExpressionParam> parent =
            expressionParamList.stream()
                .filter(p -> 0 == p.getRelationship())
                .collect(Collectors.toList());

        // 前端传参时候,存在没有父级元素,只有子级元素的情况,这时候所有子级元素升级成父级
        //    if (CollectionUtils.isEmpty(parent)) {
        //      expressionParamList.forEach(
        //          i -> {
        //            i.setRelationship(0);
        //          });
        //      parent = expressionParamList;
        //    }

        for (ExpressionParam p : parent) {
            Object express = p.getExpression();
            Object jsonPathValue = JsonPath.read(document, express.toString());
            try {
                if (!(jsonPathValue instanceof String)
                    && !(jsonPathValue instanceof Integer)
                    && !(jsonPathValue instanceof Double)
                    && !(jsonPathValue instanceof Float)
                    && !(jsonPathValue instanceof Long)) {
                    jsonPathValue = objectMapper.writeValueAsString(jsonPathValue);
                }
                parentList.add(jsonPathValue);
            } catch (JsonProcessingException e) {
                throw new BusinessException("jsonpath解析失败");
            }
        }

        Map<String, List<Object>> leafMap = new LinkedHashMap<>();
        // 子节点
        List<ExpressionParam> leaf =
            expressionParamList.stream()
                .filter(p -> 1 == p.getRelationship())
                .collect(Collectors.toList());
        for (ExpressionParam p : leaf) {
            Object name = p.getName();
            Object express = p.getExpression();
            Object jsonPathValue = JsonPath.read(document, express.toString());
            if (null == jsonPathValue) {
                log.error("解析失败");
                continue;
            }
            if (jsonPathValue instanceof List) {
                if (((List<?>) jsonPathValue).size() == 0) {
                    log.error("解析失败");
                    continue;
                }
                List l = (List) jsonPathValue;
                List<Object> leafs = new ArrayList<>();
                for (Object o : l) {
                    leafs.add(o);
                }
                leafMap.put(name.toString(), leafs);
            } else {
                log.info("子级未解析出数组, 单值为{}", jsonPathValue.toString());
                leafMap.put(name.toString(), Lists.newArrayList(jsonPathValue.toString()));
            }
        }

        List<List<Object>> leafList = new ArrayList(leafMap.values());
        List<List<Object>> results = flatten(parentList, leafList);

        if (CollectionUtils.isEmpty(results) || names.size() != results.get(0).size()) {
            log.error("title:{},value:{}", names, results);
            //      throw new BusinessException("解析失败,检查jsonpath参数或层级关系");
        }

        List<Object> nameList = new LinkedList<>();
        nameList.addAll(parent.stream().map(ExpressionParam::getName).collect(Collectors.toList()));
        nameList.addAll(leaf.stream().map(ExpressionParam::getName).collect(Collectors.toList()));

        results.add(0, nameList);
        log.info(String.valueOf(results));
        return results;
    }

    /**
     * 笛卡尔积展开表达式list
     *
     * @param lists
     * @return
     */
    @Deprecated
    private static List<List<String>> hashFlatten(List<List<String>> lists) {
        List<List<String>> resultList = new ArrayList<>();
        for (List<String> list : lists) {
            if (resultList.isEmpty()) {
                resultList.add(list);
            } else {
                resultList =
                    resultList.stream()
                        .flatMap(
                            outList ->
                                list.stream()
                                    .map(
                                        item -> {
                                            List<String> flattenedList = new ArrayList<>(outList);
                                            flattenedList.add(item);
                                            return flattenedList;
                                        }))
                        .collect(Collectors.toList());
            }
        }
        return resultList;
    }

    /**
     * @param parentNodes
     * @param leafNodes
     * @return
     */
    private static List<List<Object>> flatten(
        List<Object> parentNodes, List<List<Object>> leafNodes) {
        // align the size
        int rows = parentNodes.size();
        for (int i = 0; i < leafNodes.size(); i++) {
            rows = Math.max(leafNodes.get(i).size(), rows);
        }
        List<List<Object>> resultList = new LinkedList<>();
        for (int i = 0; i < rows; i++) {
            List<Object> items = new LinkedList<>();
            items.addAll(parentNodes);
            for (int j = 0; j < leafNodes.size(); j++) {
                List<Object> leafItems = leafNodes.get(j);
                if (leafItems.size() <= i) {
                    items.add("");
                } else {
                    items.add(leafItems.get(i));
                }
            }
            resultList.add(items);
        }

        return resultList;
    }

    /**
     * 获取运行日志
     *
     * @param logId
     * @param offset
     * @return
     */
    @Override
    public ApiLogDomain getOfflineSyncLog(Long logId, Integer offset) {
        int start = 1;
        if (offset != null && !offset.equals(0)) {
            start = offset;
        }
        ApiLogDomain response = new ApiLogDomain();
        response.setLogEnd(false);
        response.setOffset(offset);
        response.setLogContent(Lists.newArrayList());

        ApiRunningRecordDomain byId = apiRunningRecordService.getApiRunningRecord(logId);
        if (byId == null) {
            ArrayList<String> log = Lists.newArrayList("该日志未找到");
            response.setLogContent(log);
            response.setLogEnd(true);
            return response;
        }
        if (!Objects.equals(OfflineSyncTaskStatus.running.getTaskStatus(), byId.getTaskStatus())) {
            response.setLogEnd(true);
        }
        // 仅仅只是读取单个子任务的日志
        List<String> logs;
        final String dir = fileStorageConfiguration.getDir();
        String subTaskJobLogPath =
            String.format(
                "%s/log/%s/%s.log", dir, Constants.LogFileExp.OFFLINE_SYNC_API_LOG, byId.getId());

        try {
            logs = logService.read(subTaskJobLogPath, offset);
        } catch (IOException e) {
            log.error(e.getMessage());
            throw new BusinessException("读取日志异常");
        }
        if (CollectionUtils.isNotEmpty(logs)) {
            response.setOffset(logs.size() + start);
            response.setLogContent(logs);
        }
        return response;
    }

    @Override
    public void stop(Long logId) {
        throw new UnsupportedOperationException();
    }

    @Override
    public Boolean ignoreApiRunningRecord(String ids) {
        if (StringUtils.isNotBlank(ids)) {
            List<Long> res = Arrays.stream(ids.split(",")).map(e -> Long.valueOf(e))
                .collect(Collectors.toList());
            List<ApiRunningRecordEntity> apiRunningRecordEntities = apiRunningRecordService.listByIds(
                res);
            apiRunningRecordEntities.forEach(e -> {
                e.setIgnored(1);
                e.setTaskStatus(
                    OfflineDevJobStatusEnum.IGNORED.getCode());
            });
            apiRunningRecordService.updateBatchById(apiRunningRecordEntities);
        }
        return true;
    }
}
